#include <stdio.h>
#include <string.h>
#include <stdint.h>
#include <errno.h>
#include <stdlib.h>
#include <signal.h>
#include <unistd.h>
#include <sys/time.h>

#include <map>
#include <string>

#include "event.h"

#include "mw_config.h"
#include "mw_core.h"
#include "mw_channel.h"
#include "mw_process_cycle.h"
#include "mw_cycle.h"

#include <glog/logging.h>


static void mw_start_worker_processes(mw_cycle_t* cycle, int type);
static void mw_pass_open_channel(mw_channel_t *ch);
static void mw_signal_worker_processes(int signo);
static uint32_t mw_reap_children();
static void mw_master_process_exit();
static void mw_worker_process_cycle(void *data);
static void mw_worker_process_init(uint32_t priority, mw_cycle_t* cycle);
void mw_worker_process_exit();
static void mw_channel_handler(int fd, short event, void *arg);


uint32_t      mw_process;
pid_t         mw_pid;
uint32_t      mw_threaded;

sig_atomic_t  mw_reap;
sig_atomic_t  mw_sigio;
sig_atomic_t  mw_sigalrm;
sig_atomic_t  mw_terminate;
sig_atomic_t  mw_quit;
sig_atomic_t  mw_debug_quit;
uint32_t      mw_exiting;
sig_atomic_t  mw_reconfigure;
sig_atomic_t  mw_reopen;

sig_atomic_t  mw_change_binary;
pid_t         mw_new_binary;
uint32_t      mw_inherited;
uint32_t      mw_daemonized;

sig_atomic_t  mw_noaccept;
uint32_t      mw_noaccepting;
uint32_t      mw_restart;

uint64_t      cpu_affinity;
mw_cycle_t    cycle;

using namespace std;

void mw_set_callback(const string& name, const MasterEvent& event) {
    cycle.hooks[name] = event;
}

void mw_master_process_cycle(int num_processes, worker_main_fn worker_main, worker_exit_fn worker_exit, void* data) {
    uint32_t            sigio;
    sigset_t            set;
    struct itimerval    itv;
    uint32_t            live;
    uint32_t            delay;

    mw_process = MW_PROCESS_MASTER;
    if (mw_init_signals() != MW_OK) {
        exit(-1);
    }

    if (num_processes <= 0){
        num_processes = 1;
	}

    cycle.num_processes = num_processes;
    cycle.worker_main_fn = worker_main;
    cycle.worker_exit_fn = worker_exit;
    cycle.worker_data = data;

    sigemptyset(&set);
    sigaddset(&set, SIGCHLD);
    sigaddset(&set, SIGALRM);
    sigaddset(&set, SIGIO);
    sigaddset(&set, SIGINT);
    sigaddset(&set, mw_signal_value(MW_RECONFIGURE_SIGNAL));
    sigaddset(&set, mw_signal_value(MW_REOPEN_SIGNAL));
    sigaddset(&set, mw_signal_value(MW_NOACCEPT_SIGNAL));
    sigaddset(&set, mw_signal_value(MW_TERMINATE_SIGNAL));
    sigaddset(&set, mw_signal_value(MW_SHUTDOWN_SIGNAL));
    sigaddset(&set, mw_signal_value(MW_CHANGEBIN_SIGNAL));

    if (sigprocmask(SIG_BLOCK, &set, NULL) == -1) {
        LOG(ERROR) << "sigprocmask() failed";
    }

    sigemptyset(&set);

    map<string, MasterEvent>::iterator iter = cycle.hooks.find("on_start_worker");
    if (iter != cycle.hooks.end()) {
        (*iter).second();
    }

    mw_start_worker_processes(&cycle, MW_PROCESS_RESPAWN);

    mw_new_binary = 0;
    delay = 0;
    sigio = 0;
    live = 1;

    for ( ;; ) {
        if (delay > 0) {
            if (mw_sigalrm) {
                sigio = 0;
                delay *= 2;
                mw_sigalrm = 0;
            }

            //LOG(WARNING) << "termination cycle: delay=" << delay;

            itv.it_interval.tv_sec = 0;
            itv.it_interval.tv_usec = 0;
            itv.it_value.tv_sec = delay / 1000;
            itv.it_value.tv_usec = (delay % 1000 ) * 1000;

            if (setitimer(ITIMER_REAL, &itv, NULL) == -1) {
                LOG(ERROR) << "setitimer() failed";
            }
        }

        //LOG(INFO) << "sigsuspend";

        sigsuspend(&set);

        //LOG(INFO) << "wake up, sigio " << sigio;

        iter = cycle.hooks.find("on_signal");
        if (iter != cycle.hooks.end()) {
            (*iter).second();
        }

        if (mw_reap) {
            mw_reap = 0;
            live = mw_reap_children();

            iter = cycle.hooks.find("on_child");
            if (iter != cycle.hooks.end()) {
                (*iter).second();
            }
        }

        if (!live && (mw_terminate || mw_quit)) 
        {
            mw_master_process_exit();
        }

        if (mw_terminate) {
            if (delay == 0) {
                delay = 50;
            }

            if (sigio) {
                sigio--;
                continue;
            }

            sigio = cycle.num_processes;

            if (delay > 1000) {
                mw_signal_worker_processes(SIGKILL);
            } else {
                mw_signal_worker_processes(mw_signal_value(MW_TERMINATE_SIGNAL));
            }

            continue;
        }

        if (mw_quit) {
            mw_signal_worker_processes(mw_signal_value(MW_SHUTDOWN_SIGNAL));
            continue;
        }

        if (mw_reconfigure) {
            mw_reconfigure = 0;
            int ret = 0;

            iter = cycle.hooks.find("on_reload");
            if (iter != cycle.hooks.end()) {
                ret = (*iter).second();
            }
            if (ret == 0) {
                if (mw_new_binary) {
                    sleep(1);
                    mw_start_worker_processes(&cycle, MW_PROCESS_RESPAWN);
                    mw_noaccepting = 0;

                    continue;
                }

                //LOG(WARNING) << "reconfiguring";

                sleep(1);
                mw_start_worker_processes(&cycle, MW_PROCESS_JUST_RESPAWN);

                /* allow new processes to start */
                usleep(100 * 1000);

                live = 1;
                mw_signal_worker_processes(mw_signal_value(MW_SHUTDOWN_SIGNAL));
            }
        }

        if (mw_restart) {
            mw_restart = 0;
            sleep(1);
            mw_start_worker_processes(&cycle, MW_PROCESS_RESPAWN);
            live = 1;
        }

        if (mw_reopen) {
#if 0
            mw_reopen = 0;
            printf("reopening logs");
            mw_reopen_files("cloud");
            mw_signal_worker_processes(mw_signal_value(mw_reopen_SIGNAL));
#endif
        }

        if (mw_change_binary) {
#if 0
            mw_change_binary = 0;
            printf("changing binary\n");
            mw_new_binary = mw_exec_new_binary(mw_argv);
#endif
        }

        if (mw_noaccept) {
            mw_noaccept = 0;
            mw_noaccepting = 1;
            mw_signal_worker_processes(mw_signal_value(MW_SHUTDOWN_SIGNAL));
        }
    } //for ( ;; ) 
}


void mw_single_process_cycle()
{
    LOG(INFO) << "worker cycle";
    //mw_process_events_and_timers();

    if (mw_terminate || mw_quit) {
        mw_master_process_exit();
    }

    if (mw_reconfigure) {
        mw_reconfigure = 0;
        //LOG(INFO) << "reconfiguring";
    }

    if (mw_reopen) {
        mw_reopen = 0;
        //LOG(INFO) << "reopening logs";
    }
}


static void mw_start_worker_processes(mw_cycle_t* cycle_t, int type)
{
    int            i;
    mw_channel_t  ch;

    //LOG(WARNING) << "start " << cycle_t->num_processes << " worker processes type=" << type << " slot=" << mw_process_slot;

    ch.command = MW_CMD_OPEN_CHANNEL;

    for (i = 0; i < cycle_t->num_processes; i++) {
        //cpu_affinity = mw_get_cpu_affinity(i);
        mw_spawn_process(mw_worker_process_cycle, cycle_t, "worker process", type);

        ch.pid = mw_processes[mw_process_slot].pid;
        ch.slot = mw_process_slot;
        ch.fd = mw_processes[mw_process_slot].channel[0];
		//LOG(INFO)<<"mw_process_slot="<<mw_process_slot<<" pid=" <<ch.pid<<" fd="<<ch.fd;
		//LOG(INFO)<<"before mw_pass_open_channel()";
        //mw_pass_open_channel(&ch);
    }
}


static void mw_pass_open_channel(mw_channel_t *ch)
{
    int  i;

    for (i = 0; i < mw_last_process; i++) {
		//LOG(INFO)<<"mw_pass_open_channel() mw_last_process="<<mw_last_process <<" pid="<<mw_processes[i].pid <<" mw_process_slot="<<mw_process_slot <<" i="<<i;
        if (i == mw_process_slot || mw_processes[i].pid == -1 || mw_processes[i].channel[0] == -1) {
            continue;
        }
        mw_write_channel(mw_processes[i].channel[0], ch, sizeof(mw_channel_t));
    }
}


static void mw_signal_worker_processes(int signo)
{
    int      i;
    int     err;
    mw_channel_t  ch;

    switch (signo) {
        case mw_signal_value(MW_SHUTDOWN_SIGNAL):
            ch.command = MW_CMD_QUIT;
            break;

        case mw_signal_value(MW_TERMINATE_SIGNAL):
            ch.command = MW_CMD_TERMINATE;
            break;

        case mw_signal_value(MW_REOPEN_SIGNAL):
            ch.command = MW_CMD_REOPEN;
            break;

        default:
            ch.command = 0;
    }

    ch.fd = -1;

    for (i = 0; i < mw_last_process; i++) {
        //LOG(INFO) << "child: " << i << " " << mw_processes[i].pid << " e:" << mw_processes[i].exiting << " t:" << mw_processes[i].exited << " d:" << mw_processes[i].detached << " r:" <<  mw_processes[i].respawn << " j:" << mw_processes[i].just_spawn; 

        if (mw_processes[i].detached || mw_processes[i].pid == -1) {
            continue;
        }

        if (mw_processes[i].just_spawn) {
            mw_processes[i].just_spawn = 0;
            continue;
        }

        if (mw_processes[i].exiting && signo == mw_signal_value(MW_SHUTDOWN_SIGNAL)) {
            continue;
        }

        if (ch.command) {
			ch.pid = mw_processes[i].pid;
			ch.slot = i;

			//LOG(INFO)<<"mw_signal_worker_processes() slot="<<i<<" pid="<<ch.pid;
            if (mw_write_channel(mw_processes[i].channel[0], &ch, sizeof(mw_channel_t)) == MW_OK) {
                if (signo != mw_signal_value(MW_REOPEN_SIGNAL)) {
                    mw_processes[i].exiting = 1;
                }

                continue;
            }
        }

        LOG(WARNING) << "kill(" << mw_processes[i].pid << "," << signo << ")";

        if (kill(mw_processes[i].pid, signo) == -1) {
            err = errno;
            LOG(WARNING) << "kill(" << mw_processes[i].pid << "," << signo << ") failed";
            if (err == ESRCH) {
                mw_processes[i].exited = 1;
                mw_processes[i].exiting = 0;
                mw_reap = 1;
            }

            continue;
        }

        if (signo != mw_signal_value(MW_REOPEN_SIGNAL)) {
            mw_processes[i].exiting = 1;
        }
    }
}


static uint32_t mw_reap_children()
{
    int             i, n;
    uint32_t        live;
    mw_channel_t    ch;

    ch.command = MW_CMD_CLOSE_CHANNEL;
    ch.fd = -1;

    live = 0;
    for (i = 0; i < mw_last_process; i++) {
		/*
        LOG(INFO) << "child: " << i << " " << mw_processes[i].pid
            << " e:" << mw_processes[i].exiting << " t:" << mw_processes[i].exited
            << " d:" << mw_processes[i].detached << " r:" <<  mw_processes[i].respawn
            << " j:" << mw_processes[i].just_spawn;
			*/
        

        if (mw_processes[i].pid == -1) {
            continue;
        }

        if (mw_processes[i].exited) {
            if (!mw_processes[i].detached) {
                mw_close_channel(mw_processes[i].channel);

                mw_processes[i].channel[0] = -1;
                mw_processes[i].channel[1] = -1;

                ch.pid = mw_processes[i].pid;
                ch.slot = i;

                for (n = 0; n < mw_last_process; n++) {
                    if (mw_processes[n].exited || mw_processes[n].pid == -1 || mw_processes[n].channel[0] == -1) {
                        continue;
                    }

                    mw_write_channel(mw_processes[n].channel[0], &ch, sizeof(mw_channel_t));
                }
            }

            if (mw_processes[i].respawn && !mw_processes[i].exiting && !mw_terminate && !mw_quit) {
                sleep(1);
                if (mw_spawn_process(mw_processes[i].proc, mw_processes[i].data, mw_processes[i].name, i) == MW_INVALID_PID) {
                    LOG(ERROR) << "cloud not respawn " << mw_processes[i].name;
                    continue;
                }


                ch.command = MW_CMD_OPEN_CHANNEL;
                ch.pid = mw_processes[mw_process_slot].pid;
                ch.slot = mw_process_slot;
                ch.fd = mw_processes[mw_process_slot].channel[0];

                mw_pass_open_channel(&ch);

                live = 1;

                continue;
            }

            if (mw_processes[i].pid == mw_new_binary) 
            {}

            if (i == mw_last_process - 1) {
                mw_last_process--;
            } else {
                mw_processes[i].pid = -1;
            }
        } else if (mw_processes[i].exiting || !mw_processes[i].detached) {
            live = 1;
        }
    }

    return live;
}


static void mw_master_process_exit()
{
    LOG(INFO) << "master_process_exit"; 
    exit(0);
}

static void mw_worker_process_cycle(void *data)
{ 
    mw_process_t* proc = &mw_processes[mw_process_slot];
    mw_cycle_t* cycle_t = (mw_cycle_t*)data;

    mw_process = MW_PROCESS_WORKER;

    mw_worker_process_init(1, cycle_t);

    if (cycle_t->worker_main_fn != NULL) {
        cycle_t->worker_main_fn(proc->base, cycle_t->worker_data);
    }

    LOG(INFO) << mw_process_slot << " event_base_loop.";
    event_base_loop(proc->base, 0);
    LOG(INFO) << mw_process_slot << " event_base_loop exit.";
	event_base_free(proc->base);
    mw_worker_process_exit();
}


static void mw_worker_process_init(uint32_t priority, mw_cycle_t* cycle_t)
{
    sigset_t          set;
    int               n;
    //uint32_t        i;
    //struct rlimit     rlmt;

    sigemptyset(&set);

    if (sigprocmask(SIG_SETMASK, &set, NULL) == -1) {
        LOG(ERROR) << "sigprocmask() failed.";
    }

    for (n = 0; n < mw_last_process; n++) {
        if (mw_processes[n].pid == -1) {
            continue;
		}

        if (n == mw_process_slot) {
            continue;
		}

        if (mw_processes[n].channel[1] == -1) {
            continue;
		}

        if (close(mw_processes[n].channel[1]) == -1) {
            LOG(ERROR) << "close() channel failed";
        }
    }

    if (close(mw_processes[mw_process_slot].channel[0]) == -1) {
        LOG(ERROR) << "close() channel failed";
    }

#if 0
    mw_last_process = 0;
#endif

    mw_process_t* process = &mw_processes[mw_process_slot];
    process->base = event_init();
    event_set(&process->notify_event,
            process->channel[1],
            EV_READ | EV_PERSIST,
            mw_channel_handler, cycle_t);
    event_base_set(process->base, &process->notify_event);

    if(event_add(&process->notify_event, 0) == -1) {    
        LOG(ERROR) << mw_process << " can't monitor libevent notify channel";
        exit(-1);
    } 
}


void mw_worker_process_exit()
{
    LOG(WARNING) << "mw_worker_process_exit() exit";
    exit(0);
}


static void mw_channel_handler(int fd, short event, void *arg)
{
    int n;
    mw_channel_t ch;
    mw_cycle_t* cycle_t = (mw_cycle_t*)arg;

    for ( ;; ) {
        n = mw_read_channel(fd, &ch, sizeof(mw_channel_t));
        if (n == MW_ERROR) {
            LOG(ERROR) << mw_process << " read_channel error!";
            return;
        }

        if (n == MW_AGAIN) {
            return;
        }

		//LOG(INFO)<<"mw_channel_handler() read channel right";

        switch (ch.command) {
            case MW_CMD_QUIT:
                mw_quit = 1;
				//LOG(INFO)<<"mw_channel_handler() MW_CMD_QUIT";
                break;

            case MW_CMD_TERMINATE:
                mw_terminate = 1;
				//LOG(INFO)<<"mw_channel_handler() MW_CMD_TERMINATE";
                break;

            case MW_CMD_REOPEN:
                mw_reopen = 1;
				//LOG(INFO)<<"mw_channel_handler() MW_CMD_REOPEN";
                break;

            case MW_CMD_OPEN_CHANNEL:
                mw_processes[ch.slot].pid = ch.pid;
                mw_processes[ch.slot].channel[0] = ch.fd;
				//LOG(INFO)<<"mw_channel_handler() MW_CMD_OPEN_CHANNEL";
                break;

            case MW_CMD_CLOSE_CHANNEL:
                if (close(mw_processes[ch.slot].channel[0]) == -1) {
                    LOG(ERROR) << "close() channel failed";
                }

                mw_processes[ch.slot].pid = -1;
                mw_processes[ch.slot].channel[0] = -1;
				//LOG(INFO)<<"mw_channel_handler() MW_CMD_CLOSE_CHANNEL";
                break;
        }

        if (mw_quit || mw_terminate) {
			mw_process_t* process = &mw_processes[ch.slot];

			if (process) {
				LOG(INFO)<<"process->exiting="<<process->exiting;
			}
			if (process &&  !process->exiting ) {
				process->exiting = 1;
				//LOG(INFO)<<"worker_exit_fn(), ch.slot="<<ch.slot;
				if (cycle_t->worker_exit_fn != NULL)
					cycle_t->worker_exit_fn(cycle_t->worker_data);

				event_base_loopbreak(process->base);
			}
        }
    }
}
